#ifndef __M__MESSAGE_H_
#define __M__MESSAGE_H_

#include "../mqCommon/Helper.hpp"
#include "../mqCommon/Logger.hpp"
#include "../mqCommon/proto/mq_message.pb.h"
#include <iostream>
#include <unordered_map>
#include <memory>
#include <cassert>
#include <mutex>
#include <list>
#include <stack>

using namespace mq_message;
using namespace mq;
using namespace std;

/*  消息的持久化管理：
    消息以队列为单元进行持久化存储，每个队列对应一个数据文件
    不使用数据库：a.有些消息过大  b.存储消息主要是为了备份而不是查询
    以队列为单元：c.垃圾回收时，需要对文件加锁，每个队列只需加锁对应的队列文件即可，避免了激烈的锁冲突

    1.队列名
    2.源队列文件：队列名.mqd
    3.临时队列文件：队列名.mqd.tmp

    接口：
    1.消息文件的创建/删除
    2.消息的新增/删除持久化（有效标记位）
    3.历史数据恢复/垃圾回收（有效消息 <=50% && >=2000）
      回收方式：加载有效消息，写入临时文件，再改名为源文件，并需要在内存中同步新的存储位置
*/

namespace mq
{
#define DATAFILE_SUBFIX ".mqd"
#define TMPFILE_SUBFIX ".mqd.tmp"
#define VALID "1"
#define UNVALID "0"
    using m_ptr = shared_ptr<Message>;

    // 消息类：mq_message.proto文件中定义

    // 消息持久化管理类：数据文件和临时文件
    class MessageMapper
    {
    public:
        MessageMapper(const string &basedir /*消息文件创建在哪个目录下*/, const string &qname /*文件名*/)
            : _qname(qname)
        {
            string sep;
            if (basedir.back() != '/')
            {
                sep = "/";
            }
            _data_file = basedir + sep + qname + DATAFILE_SUBFIX;
            _tmp_file = basedir + sep + qname + TMPFILE_SUBFIX;

            bool ret = FileHelper::createDirectory(basedir); // 创建消息文件所在目录
            assert(ret == true);
            createMsgFile();
        }
        bool createMsgFile() // 创建数据文件
        {
            bool ret = FileHelper::createFile(_data_file);
            assert(ret == true);
            return true;
        }
        void removeMsgFile() // 删除数据文件
        {
            FileHelper fh_data(_data_file);
            fh_data.removeFile();
            FileHelper fh_tmp(_tmp_file);
            fh_tmp.removeFile();
        }
        void insert(m_ptr &msg) // 新增一条数据
        {
            insert(msg, _data_file);
        }
        void remove(m_ptr &msg) // 删除一条数据
        {
            // 1.将msg置为无效（0）
            msg->mutable_payload()->set_vaild(UNVALID);
            // 2.序列化
            string body = msg->payload().SerializeAsString();
            // 3.判断消息长度是否合法
            if (body.size() != msg->length())
            {
                DLOG("消息长度不一致，无法写入: new length:%d | old length:%d", body.size(), msg->length());
                return;
            }
            // 4.覆盖原有数据，有效变为无效
            FileHelper fh(_data_file);
            fh.write(body.c_str(), msg->offset() + 4, msg->length());
        }
        list<m_ptr> gc() // garbage collection
        {
            DLOG("garbage collection： %s[%d]", _data_file.c_str(), FileHelper::size(_data_file));
            auto result = make_shared<list<m_ptr>>();
            // 1.加载有效消息
            load(*result);
            // 2.写入临时文件
            FileHelper tmp_fh(_tmp_file);
            if (!FileHelper::exists(_tmp_file))
                tmp_fh.createFile();
            for (auto it = result->begin(); it != result->end(); it++)
            {
                m_ptr msgp = *it;
                insert(msgp, _tmp_file); // 有效消息插入临时文件
            }
            // 3.删除原源文件，临时文件改名为新源文件
            FileHelper fh_data(_data_file);
            fh_data.removeFile();
            FileHelper fh_tmp(_tmp_file);
            fh_tmp.rename(_data_file);
            DLOG("gc success： %s[%d]", _data_file.c_str(), FileHelper::size(_data_file));

            return *result;
        }
        void load(list<m_ptr> &result)
        {
            FileHelper fh(_data_file);
            size_t r_offset = 0, fsize = fh.size();
            while (r_offset < fsize)
            {
                // a.提取消息PlayLoad长度
                size_t msg_size = 0;
                fh.read((char *)&msg_size, r_offset, 4); // 4B|PlayLoad|4B|PlayLoad|4B|PlayLoad...先读取4字节消息PlayLoad长度
                r_offset += 4;
                // b.读取消息数据
                string msg_playLoad;
                msg_playLoad.resize(msg_size);
                fh.read(&msg_playLoad[0], r_offset, msg_size);
                r_offset += msg_size;
                // c.构造消息对象，反序列化
                m_ptr msgp = make_shared<Message>();
                msgp->mutable_payload()->ParseFromString(msg_playLoad);
                // d.根据是否有效，决定是否加载
                if (msgp->payload().vaild() == string(VALID))
                {
                    result.push_back(msgp);
                    // cout << "load:" << msgp->payload().properties().id() << endl;
                }
            }
        }

    private:
        void insert(m_ptr &msg, const string &file) // 向file中新增一条数据
        {
            // 1.消息序列化
            string body = msg->payload().SerializeAsString(); // 将需要存储部分序列化
            // 2.跳转至文件末尾
            FileHelper fh(file);
            size_t fsize = fh.size();
            // 3.写入文件
            size_t msg_size = body.size();
            fh.write((char *)&msg_size, fsize, 4);
            fh.write(body.c_str(), fsize + 4, body.size()); // 追加式:4B|PlayLoad|4B|PlayLoad|4B|PlayLoad...
            // 4.更新msg的存储位置&消息长度
            msg->set_offset(fsize);
            msg->set_length(body.size());
        }

    private:
        string _qname;     // 队列名
        string _data_file; // 源数据文件
        string _tmp_file;  // 临时数据文件
    };

    // 队列消息管理类：以队列为单元管理
    // 1.构造对象时：创建/打开数据文件，恢复历史消息
    // 2.新增消息/确认（删除）消息
    // 3.垃圾回收
    // 4.获取队首消息
    // 5.删除队列所有消息
    // 6.待推送/待确认/持久化消息数量
    class QueueMessage
    {
    public:
        using ptr = shared_ptr<QueueMessage>;
        QueueMessage(const string &basedir, const string &qname)
            : _mapper(basedir, qname), _qname(qname), _valid_count(0), _total_count(0) {}
        void insert(const BasicProperties *bp /*消息基本属性*/, const string &body /*实际内容*/, bool queue_durable /*是否持久化*/) // 新增消息
        {
            // 1.构造消息对象
            m_ptr msgp = make_shared<Message>();
            msgp->mutable_payload()->set_body(body);
            if (bp != nullptr)
            {
                DeliveryMode mode = queue_durable ? bp->delivery_mode() : DeliveryMode::UNDURABLE; // 队列不持久化，该消息也不持久化；队列持久化了，才根据实际需要，决定是否需要持久化
                msgp->mutable_payload()->mutable_properties()->set_id(bp->id());
                msgp->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msgp->mutable_payload()->mutable_properties()->set_routing_key(bp->routing_key());
            }
            else // 用户没有传入基本属性
            {
                DeliveryMode mode = queue_durable ? DeliveryMode::DURABLE : DeliveryMode::UNDURABLE; // 不传入属性，是否持久化默认跟随队列
                msgp->mutable_payload()->mutable_properties()->set_id(UUIDHelper::uuid());           // 生成随机id
                msgp->mutable_payload()->mutable_properties()->set_delivery_mode(mode);
                msgp->mutable_payload()->mutable_properties()->set_routing_key("");
            }
            // 2.持久化管理
            if (msgp->payload().properties().delivery_mode() == DeliveryMode::DURABLE) // 投递方式：需要持久化
            {
                msgp->mutable_payload()->set_vaild(VALID);
                {
                    unique_lock<mutex> lock(mtx);
                    _mapper.insert(msgp);
                    _durable_msgs.insert(make_pair(msgp->payload().properties().id(), msgp));
                    _valid_count++;
                    _total_count++;
                }
            }
            // 3.内存管理
            unique_lock<mutex> lock(mtx);
            _msgs.push_back(msgp);
        }
        bool remove(const string &msg_id) // 确认消息->删除消息->删除持久化->垃圾回收
        {
            unique_lock<mutex> lock(mtx);
            auto msg_it = _waitack_msgs.find(msg_id);
            if (msg_it == _waitack_msgs.end())
            {
                DLOG("未找到要删除的消息: %s", msg_id);
                return true;
            }
            else
            {
                m_ptr msgp = (*msg_it).second;
                _waitack_msgs.erase(msg_id);
                if (msgp->payload().properties().delivery_mode() == DeliveryMode::DURABLE) // 投递方式：持久化
                {
                    _mapper.remove(msgp);
                    _durable_msgs.erase(msgp->payload().properties().id());
                    _valid_count--;
                    gc(); // 垃圾回收
                }
            }
        }
        m_ptr front() // 获取待推送链表的队首消息，用于推送
        {
            unique_lock<mutex> lock(mtx);
            // 取出队首消息
            if (_msgs.size() == 0) // 没有消息可推送
                return m_ptr();
            m_ptr msgp = _msgs.front();
            _msgs.pop_front();
            // 放入待确认map，防止消息丢失
            _waitack_msgs.insert(make_pair(msgp->payload().properties().id(), msgp));
            return msgp;
        }
        void clear() // 清理所有队列数据
        {
            unique_lock<mutex> lock(mtx);
            _mapper.removeMsgFile();
            _msgs.clear();
            _waitack_msgs.clear();
            _durable_msgs.clear();
            _valid_count = 0;
            _total_count = 0;
        }
        void recovery()
        {
            unique_lock<mutex> lock(mtx);
            _mapper.load(_msgs); // 恢复历史消息
            for (auto it = _msgs.begin(); it != _msgs.end(); it++)
            {
                m_ptr msgp = *it;
                _durable_msgs.insert(make_pair(msgp->payload().properties().id(), msgp)); // 更新内存中消息存储位置
            }
            _valid_count = _total_count = _msgs.size();
        }
        size_t getable_count() // 可推送消息数量
        {
            unique_lock<mutex> lock(mtx);
            return _msgs.size();
        }
        size_t waitack_count() // 待确认消息数量
        {
            unique_lock<mutex> lock(mtx);
            return _waitack_msgs.size();
        }
        size_t total_count() // 总的持久化消息数量
        {
            unique_lock<mutex> lock(mtx);
            return _total_count;
        }
        size_t vaild_count() // 有效持久化消息数量
        {
            unique_lock<mutex> lock(mtx);
            return _valid_count;
        }

    private:
        bool gcCheck() // 判断是否需要垃圾回收
        {
            return (_valid_count >= 2000) && (_valid_count * 1.0 / _total_count * 100 <= 50);
        }
        void gc() // 垃圾回收
        {
            if (!gcCheck())
                return;
            auto msgs = _mapper.gc(); // 有效消息链表
            if (msgs.size() != _durable_msgs.size())
            {
                ELOG("消息存储有问题");
                return;
            }
            _durable_msgs.clear();
            for (auto it = msgs.begin(); it != msgs.end(); it++)
            {
                m_ptr msgp = *it;
                _durable_msgs.insert(make_pair(msgp->payload().properties().id(), msgp)); // 更新内存中消息存储位置
            }
            _valid_count = _total_count = _durable_msgs.size();
        }

    private:
        mutex mtx;     // 对公有成员访问需要加锁
        string _qname; // 队列名称
        size_t _valid_count;
        size_t _total_count;                        // 垃圾回收：持久化的消息中->有效消息数量/总的消息数量
        MessageMapper _mapper;                      // 持久化管理句柄
        list<m_ptr> _msgs;                          // 待推送消息链表
        unordered_map<string, m_ptr> _waitack_msgs; // 待确认消息map
        unordered_map<string, m_ptr> _durable_msgs; // 持久化消息map
    };

    // 总体消息管理类：管理每一个队列消息
    // 1.队列消息管理句柄的初始化/销毁:恢复历史消息
    // 2.队列消息操作:队列消息的增删查
    class MessageManager
    {
    public:
        using manager_ptr = shared_ptr<MessageManager>;
        MessageManager(const string &basedir) : _basedir(basedir) {}
        void initQueueMessage(const string &qname) // 初始化队列消息管理句柄
        {
            QueueMessage::ptr qmp;
            {
                unique_lock<mutex> lock(mtx);
                auto it = _queue_msgs.find(qname);
                if (it != _queue_msgs.end()) // 句柄已经存在
                    return;
                qmp = make_shared<QueueMessage>(_basedir, qname); // 不存在，创建句柄
                _queue_msgs.insert(make_pair(qname, qmp));
            }
            // 不在加锁的代码块的构造函数内进行历史消息恢复，将恢复数据独立独立在锁外
            qmp->recovery();
        }
        void destroyQueueMessage(const string &qname) // 销毁队列消息管理句柄
        {
            QueueMessage::ptr qmp;
            {
                unique_lock<mutex> lock(mtx);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) // 句柄已经销毁
                    return;
                qmp = it->second;
                _queue_msgs.erase(it);
            }
            // 不在加锁的代码块的构造函数内进行队列消息销毁，将恢复数据独立独立在锁外
            qmp->clear();
        }
        bool insert(const string &qname, const BasicProperties *bp, const string &body, bool queue_durable) // 向指定队列新增一条消息
        {
            QueueMessage::ptr qmp;
            {
                unique_lock<mutex> lock(mtx);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) // 找不到队列
                {
                    DLOG("找不到该消息对应的队列: %s", qname.c_str());
                    return false;
                }
                qmp = it->second;
            }
            qmp->insert(bp, body, queue_durable);
            return true;
        }
        m_ptr front(const string &qname) // 获取指定队列的队首消息，进行推送
        {
            QueueMessage::ptr qmp;
            {
                unique_lock<mutex> lock(mtx);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) // 找不到队列
                {
                    DLOG("找不到队列，无法获取队首消息: %s", qname.c_str());
                    return m_ptr();
                }
                qmp = it->second;
            }

            return qmp->front();
        }
        void ack(const string &qname, const string &msg_id) // 对指定队列中的消息进行确认
        {
            QueueMessage::ptr qmp;
            {
                unique_lock<mutex> lock(mtx);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) // 找不到队列
                {
                    DLOG("找不到队列，无法确认消息: %s ack %s", qname.c_str(), msg_id.c_str());
                    return;
                }
                qmp = it->second;
            }
            qmp->remove(msg_id);
        }
        void clear() // 销毁所有队列数据
        {
            stack<QueueMessage::ptr> clear_st;
            {
                unique_lock<mutex> lock(mtx);
                for (auto it = _queue_msgs.begin(); it != _queue_msgs.end(); it++)
                {
                    clear_st.push(it->second); // 入清理栈
                }
                _queue_msgs.clear();
                _basedir.clear();
            }
            // 将所有队列消息管理句柄取出后，统一清理
            while (!clear_st.empty())
            {
                QueueMessage::ptr qmp = clear_st.top();
                clear_st.pop();
                qmp->clear();
            }
        }
        // 获取指定队列的消息数量
        size_t getable_count(const string &qname) // 待推送消息数量
        {
            QueueMessage::ptr qmp;
            {
                unique_lock<mutex> lock(mtx);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) // 找不到队列
                {
                    DLOG("找不到队列，无法获取指定队列的待推送消息数量: %s", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->getable_count();
        }
        size_t waitack_count(const string &qname) // 待确认消息数量
        {
            QueueMessage::ptr qmp;
            {
                unique_lock<mutex> lock(mtx);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) // 找不到队列
                {
                    DLOG("找不到队列，无法获取指定队列的待推送消息数量: %s", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->waitack_count();
        }
        size_t total_count(const string &qname) // 总的持久化消息数量
        {
            QueueMessage::ptr qmp;
            {
                unique_lock<mutex> lock(mtx);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) // 找不到队列
                {
                    DLOG("找不到队列，无法获取指定队列的待推送消息数量: %s", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->total_count();
        }
        size_t vaild_count(const string &qname) // 有效持久化消息数量
        {
            QueueMessage::ptr qmp;
            {
                unique_lock<mutex> lock(mtx);
                auto it = _queue_msgs.find(qname);
                if (it == _queue_msgs.end()) // 找不到队列
                {
                    DLOG("找不到队列，无法获取指定队列的待推送消息数量: %s", qname.c_str());
                    return 0;
                }
                qmp = it->second;
            }
            return qmp->vaild_count();
        }

    private:
    private:
        mutex mtx;                                            // 锁只需要保护_queue_msgs即可,队列消息管理内部自己保证线程安全
        string _basedir;                                      // 所有队列文件的存放目录
        unordered_map<string, QueueMessage::ptr> _queue_msgs; // 每个队列的消息管理句柄：map[qname,QueueMessage]
    };
}

#endif